package com.edu.realtime.app.dws;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.app.func.DimAsyncFunction;
import com.edu.realtime.bean.TestQuestionAnswer;
import com.edu.realtime.util.DateFormatUtil;
import com.edu.realtime.util.MyClickhouseUtil;
import com.edu.realtime.util.MyKafkaUtil;

/**
 * Created on 2022/10/22.
 *
 * @author Topus
 * @desc 考试域：题目粒度答题次数正确答题次数及独立用户数
 */
public class DwsTestQuestionAnswerWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        /*
        需要：题目粒度 答题正确次数以及答题次数 答对题目的独立用户数 答题的独立用户数
            针对独立用户数使用mapState的状态编程：如果答题了put<user_id,0>如果答对题目了put<user_id,1>
         */
        // 从kafka中读取数据
        String topic = "dwd_test_exam_do_paper";
        String groupId = "dws_test_question_answer_window_group";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));
        SingleOutputStreamOperator<JSONObject> mapDS = source.map(JSONObject::parseObject);
        // 水位线设置
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = mapDS
            .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000L;
                    }
                }));
        // keyBy
        KeyedStream<JSONObject, String> keyByDS = withWatermarkDS.keyBy(jsonObj -> jsonObj.getString("question_id"));
        // 状态编程求出数据
        SingleOutputStreamOperator<TestQuestionAnswer> processDS =
            keyByDS.process(new KeyedProcessFunction<String, JSONObject, TestQuestionAnswer>() {
                // 状态编程求出来答题相关独立人数用户做过了<user_id,false> 如果题目还做对了<user_id,true>
                private MapState<String, Boolean> mapState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    MapStateDescriptor<String, Boolean> stateDescriptor =
                        new MapStateDescriptor<>("test-question-answer-state", String.class, Boolean.class);
                    stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    mapState = getRuntimeContext().getMapState(stateDescriptor);
                }

                @Override
                public void processElement(JSONObject value,
                    KeyedProcessFunction<String, JSONObject, TestQuestionAnswer>.Context ctx,
                    Collector<TestQuestionAnswer> out) throws Exception {

                    // 状态编程求出做题独立用户以及做题独立用户数
                    String questionId = value.getString("question_id");
                    String userId = value.getString("user_id");
                    Integer isCorrect = value.getInteger("is_correct");
                    long correctCount = 0L;
                    if (isCorrect == 1) {
                        correctCount = 1L;
                    }

                    long userCount = 0L;
                    long correctUserCount = 0L;
                    //
                    if (mapState.contains(userId)) {
                        // 如果userId存在，判断题目是否正确
                        if (correctCount == 1) {
                            // 如果题目正确，判断map中是否已经为1了
                            Boolean flag = mapState.get(userId);
                            // 如果是第一次做对题目
                            if (!flag) {
                                mapState.put(userId, true);
                                correctUserCount = 1L;
                            }
                        }
                    } else {
                        // 如果该用户第一次答题
                        userCount = 1L;
                        if (isCorrect == 1) {
                            correctUserCount = 1L;
                            mapState.put(userId, true);
                        } else {
                            mapState.put(userId, false);
                        }
                    }

                    TestQuestionAnswer testQuestionAnswer = TestQuestionAnswer.builder()
                        //
                        .questionId(questionId)
                        //
                        .answerCount(1L)
                        //
                        .correctAnswerCount(correctCount)
                        //
                        .answerUserCount(userCount)
                        //
                        .correctAnswerUserCount(correctUserCount).build();

                    out.collect(testQuestionAnswer);
                }
            });

        // 开窗
        AllWindowedStream<TestQuestionAnswer, TimeWindow> windowDS = processDS
            .windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        // 聚合
        SingleOutputStreamOperator<TestQuestionAnswer> reduceDS =
            windowDS.reduce(new ReduceFunction<TestQuestionAnswer>() {
                @Override
                public TestQuestionAnswer reduce(TestQuestionAnswer value1, TestQuestionAnswer value2)
                    throws Exception {
                    value1.setAnswerCount(value1.getAnswerCount() + value2.getAnswerCount());
                    value1.setAnswerUserCount(value1.getAnswerUserCount() + value2.getAnswerUserCount());
                    value1.setCorrectAnswerCount(value1.getCorrectAnswerCount() + value2.getCorrectAnswerCount());
                    value1.setCorrectAnswerUserCount(
                        value1.getCorrectAnswerUserCount() + value2.getCorrectAnswerUserCount());
                    return value1;
                }
            }, new AllWindowFunction<TestQuestionAnswer, TestQuestionAnswer, TimeWindow>() {
                @Override
                public void apply(TimeWindow window, Iterable<TestQuestionAnswer> values,
                    Collector<TestQuestionAnswer> out) throws Exception {
                    String sst = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());

                    for (TestQuestionAnswer value : values) {
                        value.setStt(sst);
                        value.setEdt(edt);
                        value.setTs(System.currentTimeMillis());
                        //
                        out.collect(value);
                    }
                }
            });

        // join维度
        SingleOutputStreamOperator<TestQuestionAnswer> joinDS =
            AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TestQuestionAnswer>("dim_test_question_info") {
                @Override
                public void join(TestQuestionAnswer input, JSONObject dimInfo) {
                    String questionTxt = dimInfo.getString("QUESTION_TXT").replaceAll("<[^>]*>", "");
                    input.setQuestionName(questionTxt);
                }

                @Override
                public String getKey(TestQuestionAnswer obj) {
                    return obj.getQuestionId();
                }
            }, 120, TimeUnit.SECONDS);

        // 写入clickhouse
        joinDS.addSink(MyClickhouseUtil.<TestQuestionAnswer>getSinkFunction(
            "insert into dws_test_question_answer_window values(?,?,?,?,?,?,?,?,?)"));

        //
        env.execute();
    }
}